#pragma once

#include <chrono>
#include <functional>
#include <memory>
#include <string>

#include <boost/property_tree/json_parser.hpp>
#include <boost/property_tree/ptree.hpp>

#include "logger.hpp"
#include "mysql_outbox.hpp"
#include "outbox_sync_worker.hpp"
#include "data_redis.hpp"
#include "data_es.hpp"
#include "cache_service.hpp"

namespace suye
{
class UserOutboxDispatcher
{
public:
    using ptr = std::shared_ptr<UserOutboxDispatcher>;

    UserOutboxDispatcher(const OutboxTable::ptr& outbox_table,
                         const std::shared_ptr<ESUser>& es_user,
                         const CacheService::ptr& cache_service)
        : _outbox_table(outbox_table)
        , _es_user(es_user)
        , _cache_service(cache_service)
    {
        if (!_outbox_table)
        {
            throw std::invalid_argument("UserOutboxDispatcher requires valid OutboxTable instance");
        }
        _worker = std::make_shared<OutboxSyncWorker>(_outbox_table, 64, std::chrono::milliseconds(500));
        register_handlers();
    }

    OutboxSyncWorker::ptr worker() const
    {
        return _worker;
    }

    void start()
    {
        if (_worker)
        {
            _worker->start();
        }
    }

    void stop()
    {
        if (_worker)
        {
            _worker->stop();
        }
    }

private:
    void register_handlers()
    {
        if (_worker)
        {
            _worker->register_handler("USER_CREATED", [this](const std::shared_ptr<OutboxEvent>& event, std::string& err) {
                return this->handle_user_created(event, err);
            });

            _worker->register_handler("USER_PROFILE_UPDATED", [this](const std::shared_ptr<OutboxEvent>& event, std::string& err) {
                return this->handle_user_profile_updated(event, err);
            });
        }
    }

    bool handle_user_created(const std::shared_ptr<OutboxEvent>& event, std::string& err)
    {
        auto payload_tree = parse_payload(event, err);
        if (!payload_tree)
        {
            return false;
        }

        const std::string user_id = payload_tree->get<std::string>("user_id", "");
        const std::string nickname = payload_tree->get<std::string>("nickname", "");
        const std::string phone = payload_tree->get<std::string>("phone", "");
        const std::string source = payload_tree->get<std::string>("source", "");

        LOG_INFO("OutboxDispatcher 处理 USER_CREATED: uid={}, nickname={}, phone={}, source={}", user_id, nickname, phone, source);

        // 清理可能存在的缓存
        if (_cache_service)
        {
            const std::string cache_key = "user:profile:" + user_id;
            _cache_service->invalidate(cache_key);
        }

        if (_es_user)
        {
            bool es_ok = _es_user->appendData(user_id, phone, nickname, "", "");
            if (!es_ok)
            {
                err = "ES appendData failed";
                return false;
            }
        }

        return true;
    }

    bool handle_user_profile_updated(const std::shared_ptr<OutboxEvent>& event, std::string& err)
    {
        auto payload_tree = parse_payload(event, err);
        if (!payload_tree)
        {
            return false;
        }

        const std::string user_id = payload_tree->get<std::string>("user_id", "");
        const std::string field = payload_tree->get<std::string>("field", "");
        const std::string value = payload_tree->get<std::string>("new", "");

        LOG_INFO("OutboxDispatcher 处理 USER_PROFILE_UPDATED: uid={}, field={}, value={}", user_id, field, value);

        // 延迟双删策略：先删除缓存
        if (_cache_service)
        {
            const std::string cache_key = "user:profile:" + user_id;
            _cache_service->delayed_double_delete(cache_key, std::chrono::milliseconds(500));
            LOG_DEBUG("OutboxDispatcher 触发延迟双删: {}", cache_key);
        }

        if (_es_user)
        {
            // ES 暂时只支持全量 upsert，后续可增加增量更新接口
            LOG_DEBUG("OutboxDispatcher 暂时跳过 ES 增量更新，等待完善 appendData 或新增 update 接口");
        }

        return true;
    }

    std::unique_ptr<boost::property_tree::ptree> parse_payload(const std::shared_ptr<OutboxEvent>& event, std::string& err)
    {
        if (!event)
        {
            err = "event is null";
            return nullptr;
        }

        try
        {
            std::stringstream ss(event->payload());
            auto tree = std::make_unique<boost::property_tree::ptree>();
            boost::property_tree::read_json(ss, *tree);
            return tree;
        }
        catch (const std::exception& e)
        {
            err = e.what();
            return nullptr;
        }
    }

private:
    OutboxTable::ptr _outbox_table;
    std::shared_ptr<ESUser> _es_user;
    CacheService::ptr _cache_service;

    std::shared_ptr<OutboxSyncWorker> _worker;
};
}
